-- **************************************************************************
--    Project Name:   dwd层-网点中心综合账单 清洗
--    Job Name:       spmi_dwd.dwd_spmn_apack_bill_dt（分区字段：账单产生时间）
--    Description :   从ods层 spmn_apack_bill 表取每天T-1增量更新的数据，T-1 更新的数据划分为三部分，A： 最近30天账单产生的数据， B：最近31-60天账单产生的数据，C：60天之前账单产生的数据，
--                    A部分数据和 spmi_dwd.dwd_spmn_apack_bill_dt 最近30天基于运单号去重，然后回刷回0-30天分区；
--                    B部分数据和  spmi_dwd.dwd_spmn_apack_bill_dt 最近30-60天数据基于运单号去重，然后回刷回31-60天分区
--                    C部分数据直接丢弃，spmi_dwd.dwd_spmn_apack_bill_dt 60天之前的 数据不予更新
--    Author :        songjun
--    date：          2022/5/23
-- **************************************************************************
--    modify by songjun 2022/05/23 上线
-- **************************************************************************

SET hive.merge.mapfiles = true; --  默认true，在map-only任务结束时合并小文件
SET hive.merge.mapredfiles = true; -- 默认false，在map-reduce任务结束时合并小文件
SET hive.merge.size.per.task = 25610001000; -- 默认256M
SET hive.merge.smallfiles.avgsize = 16777216; -- 当输出文件的平均大小小于16m该值时，启动一个独立的map-reduce任务进行文件merge

with tmp_spmi_apack_bill_incr as (
select
id
,waybill_no
,package_no
,package_type
,bill_no
,operation_network_id
,operation_network_code
,operation_network_name
,operation_financial_center_id
,operation_financial_center_name
,operation_financial_center_code
,send_network_id
,send_network_code
,send_network_name
,send_financial_center_id
,send_financial_center_name
,send_financial_center_code
,fee_type_id
,fee_type_name
,fee_type_code
,product_type_id
,product_type_name
,product_type_code
,goods_type_id
,goods_type_name
,goods_type_code
,waybill_piece
,waybill_num
,begin_id
,begin_name
,begin_code
,begin_city_id
,begin_city_name
,begin_province_id
,begin_province_name
,destination_id
,destination_code
,destination_name
,destination_city_id
,destination_city_name
,destination_province_id
,destination_province_name
,destination_financial_center_id
,destination_financial_center_name
,destination_financial_center_code
,sign_network_id
,sign_network_code
,sign_network_name
,is_sign
,sign_time
,bill_generation_time
,bill_generation_date
,bill_update_time
,business_happens_time
,waybill_operation_time
,dec_network_id
,dec_network_code
,dec_network_name
,dec_financial_center_id
,dec_financial_center_code
,dec_financial_center_name
,fee_cost
,fee_before_collect
,cal_cost_desc
,remarks
,is_not_finish
,is_costint
,collect_no
,is_payment
,payment_time
,waybill_weight
,package_receipt_weight
,package_charge_weight
,settle_weight
,is_confirmeder_bill
,confirmeder_network_id
,confirmeder_network_code
,confirmeder_network_name
,confirmeder_id
,confirmeder_name
,confirmeder_time
,is_verifier
,verifier_id
,verifier_name
,verifier_time
,is_retry
,is_bad_piece
,is_rejection
,is_lock
,lock_by
,lock_by_name
,lock_time
,is_enable
,is_delete
,create_by
,update_by
,create_by_name
,update_by_name
,create_time
,update_time
,version
,last_update_time_sync
,trace_id
,unpack_network_id
,unpack_network_name
,unpack_network_code
,unpack_financial_center_id
,unpack_financial_center_name
,unpack_financial_center_code
,unpack_time
,unpack_current_time
,pack_network_id
,pack_network_name
,pack_network_code
,is_delivery
,operation_parent_network_id
,operation_parent_network_name
,operation_parent_network_code
,cast(to_date(bill_generation_date) as string) as dt
from spmi_ods.spmn_apack_bill where dt = '{{ execution_date | cst_ds }}'
),
tmp_spmi_apack_bill_incr_A as (
select * from tmp_spmi_apack_bill_incr where to_date(bill_generation_date) between date_add('{{ execution_date | cst_ds }}',-30) and '{{ execution_date | cst_ds }}'
),
tmp_spmi_apack_bill_incr_B as (
select * from tmp_spmi_apack_bill_incr where to_date(bill_generation_date) between date_add('{{ execution_date | cst_ds }}',-60) and date_add('{{ execution_date | cst_ds }}',-31)
),
tmp_dwd_spmi_apack_bill_incr_dt as (
select
id
,waybill_no
,package_no
,package_type
,bill_no
,operation_network_id
,operation_network_code
,operation_network_name
,operation_financial_center_id
,operation_financial_center_name
,operation_financial_center_code
,send_network_id
,send_network_code
,send_network_name
,send_financial_center_id
,send_financial_center_name
,send_financial_center_code
,fee_type_id
,fee_type_name
,fee_type_code
,product_type_id
,product_type_name
,product_type_code
,goods_type_id
,goods_type_name
,goods_type_code
,waybill_piece
,waybill_num
,begin_id
,begin_name
,begin_code
,begin_city_id
,begin_city_name
,begin_province_id
,begin_province_name
,destination_id
,destination_code
,destination_name
,destination_city_id
,destination_city_name
,destination_province_id
,destination_province_name
,destination_financial_center_id
,destination_financial_center_name
,destination_financial_center_code
,sign_network_id
,sign_network_code
,sign_network_name
,is_sign
,sign_time
,bill_generation_time
,bill_generation_date
,bill_update_time
,business_happens_time
,waybill_operation_time
,dec_network_id
,dec_network_code
,dec_network_name
,dec_financial_center_id
,dec_financial_center_code
,dec_financial_center_name
,fee_cost
,fee_before_collect
,cal_cost_desc
,remarks
,is_not_finish
,is_costint
,collect_no
,is_payment
,payment_time
,waybill_weight
,package_receipt_weight
,package_charge_weight
,settle_weight
,is_confirmeder_bill
,confirmeder_network_id
,confirmeder_network_code
,confirmeder_network_name
,confirmeder_id
,confirmeder_name
,confirmeder_time
,is_verifier
,verifier_id
,verifier_name
,verifier_time
,is_retry
,is_bad_piece
,is_rejection
,is_lock
,lock_by
,lock_by_name
,lock_time
,is_enable
,is_delete
,create_by
,update_by
,create_by_name
,update_by_name
,create_time
,update_time
,version
,last_update_time_sync
,trace_id
,unpack_network_id
,unpack_network_name
,unpack_network_code
,unpack_financial_center_id
,unpack_financial_center_name
,unpack_financial_center_code
,unpack_time
,unpack_current_time
,pack_network_id
,pack_network_name
,pack_network_code
,is_delivery
,operation_parent_network_id
,operation_parent_network_name
,operation_parent_network_code
,dt
from spmi_dwd.dwd_spmn_apack_bill_dt where dt between date_add('{{ execution_date | cst_ds }}',-60) and  '{{ execution_date | cst_ds }}'
)

insert overwrite table spmi_dwd.dwd_spmn_apack_bill_dt partition(dt)
select
id
,waybill_no
,package_no
,package_type
,bill_no
,operation_network_id
,operation_network_code
,operation_network_name
,operation_financial_center_id
,operation_financial_center_name
,operation_financial_center_code
,send_network_id
,send_network_code
,send_network_name
,send_financial_center_id
,send_financial_center_name
,send_financial_center_code
,fee_type_id
,fee_type_name
,fee_type_code
,product_type_id
,product_type_name
,product_type_code
,goods_type_id
,goods_type_name
,goods_type_code
,waybill_piece
,waybill_num
,begin_id
,begin_name
,begin_code
,begin_city_id
,begin_city_name
,begin_province_id
,begin_province_name
,destination_id
,destination_code
,destination_name
,destination_city_id
,destination_city_name
,destination_province_id
,destination_province_name
,destination_financial_center_id
,destination_financial_center_name
,destination_financial_center_code
,sign_network_id
,sign_network_code
,sign_network_name
,is_sign
,sign_time
,bill_generation_time
,bill_generation_date
,bill_update_time
,business_happens_time
,waybill_operation_time
,dec_network_id
,dec_network_code
,dec_network_name
,dec_financial_center_id
,dec_financial_center_code
,dec_financial_center_name
,fee_cost
,fee_before_collect
,cal_cost_desc
,remarks
,is_not_finish
,is_costint
,collect_no
,is_payment
,payment_time
,waybill_weight
,package_receipt_weight
,package_charge_weight
,settle_weight
,is_confirmeder_bill
,confirmeder_network_id
,confirmeder_network_code
,confirmeder_network_name
,confirmeder_id
,confirmeder_name
,confirmeder_time
,is_verifier
,verifier_id
,verifier_name
,verifier_time
,is_retry
,is_bad_piece
,is_rejection
,is_lock
,lock_by
,lock_by_name
,lock_time
,is_enable
,is_delete
,create_by
,update_by
,create_by_name
,update_by_name
,create_time
,update_time
,version
,last_update_time_sync
,trace_id
,unpack_network_id
,unpack_network_name
,unpack_network_code
,unpack_financial_center_id
,unpack_financial_center_name
,unpack_financial_center_code
,unpack_time
,unpack_current_time
,pack_network_id
,pack_network_name
,pack_network_code
,is_delivery
,operation_parent_network_id
,operation_parent_network_name
,operation_parent_network_code
 ,dt
 from (
select * from
(
select * , row_number() over(partition by waybill_no,fee_type_id order by  update_time desc) as rnk  from
(select
* from tmp_dwd_spmi_apack_bill_incr_dt where to_date(bill_generation_date) between date_add('{{ execution_date | cst_ds }}',-30) and '{{ execution_date | cst_ds }}'
union all
select
* from tmp_spmi_apack_bill_incr_A
) tmp_union  ) tmp_apack_rnk where  rnk  = 1

union all
select * from
(
select * , row_number() over(partition by waybill_no ,fee_type_id order by  update_time desc) as rnk  from
(select
* from tmp_dwd_spmi_apack_bill_incr_dt where to_date(bill_generation_date) between date_add('{{ execution_date | cst_ds }}',-60) and date_add('{{ execution_date | cst_ds }}',-31)
union all
select
* from tmp_spmi_apack_bill_incr_B
) tmp_union  ) tmp_apack_rnk where  rnk  = 1

) tmp_apack_all
distribute by dt, abs(hash(id)) % 30;